package com.lagou.client;

import com.lagou.handler.UserClientHandler;
import com.lagou.zookeeper.GetNodeData;
import com.lagou.zookeeper.ZkConstant;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 消费者
 */
public class RPCConsumer {

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime()
                                                .availableProcessors());

    //2.声明一个自定义事件处理器  UserClientHandler
    private UserClientHandler userClientHandler;
    private ChannelFuture future;
    private Bootstrap bootstrap;

    private String ip;
    private int port;


    //3.编写方法,初始化客户端  ( 创建连接池  bootStrap  设置bootstrap  连接服务器)
    public void initClient(String ip, int port) throws InterruptedException, KeeperException {
        this.ip = ip;
        this.port = port;
        //1) 初始化UserClientHandler
        userClientHandler = new UserClientHandler();
        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)创建客户端的引导对象
        bootstrap = new Bootstrap();
        //4)配置启动引导对象
        bootstrap.group(group)
                 //设置通道为NIO
                 .channel(NioSocketChannel.class)
                 //设置请求协议为TCP
                 .option(ChannelOption.TCP_NODELAY, true)
                 //监听channel 并初始化
                 .handler(new ChannelInitializer<SocketChannel>() {
                     protected void initChannel(SocketChannel socketChannel) throws Exception {
                         //获取ChannelPipeline
                         ChannelPipeline pipeline = socketChannel.pipeline();
                         //设置编码
                         pipeline.addLast(new StringEncoder());
                         pipeline.addLast(new StringDecoder());
                         //添加自定义事件处理器
                         pipeline.addLast(userClientHandler);
                     }
                 });

        Thread.sleep(2000);
        //5)连接服务端
        future = bootstrap.connect(ip, port)
                          .sync();
        if (future.isSuccess()) {
            System.out.println("客户端连接服务器" + ip + ":" + port + "成功!!!");
        }

    }

    //4.编写一个方法,使用JDK的动态代理创建对象
    // serviceClass 接口类型,根据哪个接口生成子类代理对象;   providerParam :  "UserService#sayHello#"
    public Object createProxy(Class<?> serviceClass, final String providerParam) {
        return Proxy.newProxyInstance(Thread.currentThread()
                                            .getContextClassLoader(),
                                      new Class[]{serviceClass}, (Object o, Method method, Object[] objects) -> {
                    System.out.println("》》》》》》》》请求的服务器节点地址" + ip + ":" + port);
                    long startTime = System.currentTimeMillis();

                    //2)给UserClientHandler 设置param参数
                    userClientHandler.setParam(providerParam + objects[0]);

                    //3).使用线程池,开启一个线程处理处理call() 写操作,并返回结果
                    Object result = null;
                    try {
                        result = executorService.submit(userClientHandler)
                                                       .get();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                    long costTime = System.currentTimeMillis() - startTime;
                    //更新zk接口数据
                    updateServerNodeData(ip, port, costTime);
                    //4)return 结果
                    return result;
                });
    }

    private void updateServerNodeData(String ip, int port, long responseTime) {
        ZooKeeper zooKeeper = GetNodeData.zooKeeper;
        try {
            List<String> children = zooKeeper.getChildren(ZkConstant.ROOT_NODE, true);
            for (String child : children) {
                if ((ip + ":" + port).endsWith(child)) {
                    zooKeeper.setData(ZkConstant.ROOT_NODE + "/" + child, (responseTime + "").getBytes("UTF-8"), -1);
                    System.out.println("更新节点" + child + "数据成功");
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }

    public void disconnect(String address) {
        System.out.println("与服务器" + address + "断开连接");
        ChannelFuture cf = future.channel().closeFuture();
        ChannelFuture cf2 = future.channel().close();
        if (cf.isSuccess()) {
            System.out.println("--------------------");
        }
        if (cf2.isSuccess()) {
            System.out.println("====================");
        }
        System.out.println("与服务器" + address + "断开连接成功");
    }

}
